package TestSparkStreaming;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;

/**
 * 该算子的作用是对当前窗口内的数据进行计数，返回一个DStream对象。下图演示了“每隔6秒输出最近3000秒接收的字符串个数”
 */
public class TestCountByWindow {
    public static final String checkpointDir = "checkpoint_dir/";
    public static void main(String[] args) throws InterruptedException {
        SparkConf sparkConf = new SparkConf().setAppName("TestCountByWindow").setMaster("local[2]");

        JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);

        JavaStreamingContext javaStreamingContext = new JavaStreamingContext(javaSparkContext, new Duration(1000));

        javaSparkContext.checkpointFile(checkpointDir);

        JavaReceiverInputDStream<String> javaReceiverInputDStream = javaStreamingContext.socketTextStream("localhost", 9999);

        JavaDStream<Long> countByWindow = javaReceiverInputDStream.countByWindow(new Duration(1000 * 3000), new Duration(1000 * 6));

        countByWindow.print();

        javaStreamingContext.start();

        javaStreamingContext.awaitTermination();
    }
}
